/*
 * =====================================================================================
 *
 *       Filename:  cl_tcp_conn.cxx
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  03/29/2011 03:24:29 PM
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Ken Chen (Ken), chen_kenz@emc.com
 *        Company:  EMC
 *
 * =====================================================================================
 */

#ifndef __GXX_EXPERIMENTAL_CXX0X__
#define __GXX_EXPERIMENTAL_CXX0X__
#endif 

#include <cstring>
#include <cassert>
#include "cl_tcp_conn.hxx"
#include "cl_socket.hxx"
#include "cl_channel.hxx"
#include "cl_event_looper.hxx" 
#include "cl_socket_addr.hxx"
#include "cl_logger.hxx"
#include "cl_buffer.hxx" 
#include "cl_common_defs.hxx"

namespace neo
{
using namespace std;

CLTcpConn::CLTcpConn(CLEventLooper* ioLooper,
                     int connId,
                     int connfd, 
                     CLSocketAddrUPtr peerAddr) 
	: m_ioLooper(ioLooper),
	m_connId(connId),
	m_socket(new CLSocket(connfd, move(peerAddr))), 
  m_ioChannel(new CLChannel(connfd)),
  m_curState(TCP_CONNECTING) 
{
  m_ioChannel->setReadCallback(bind(&CLTcpConn::handleRead, this));
  m_ioChannel->setWriteCallback(bind(&CLTcpConn::handleWrite, this));
  m_ioChannel->setCloseCallback(bind(&CLTcpConn::handleClose, this));
  m_ioChannel->setErrorCallback(bind(&CLTcpConn::handleError, this));
	CLLOG_D("NET", "%s: %d: %d", "CLTcpConn ctor", connId, connfd);
}

CLTcpConn::~CLTcpConn() 
{
  CLLOG_D("NET", "CLTcpConn dtor: %d", m_ioChannel->fd());
}

bool CLTcpConn::shutDownWrite(void) 
{
	return m_socket->shutDown(CL_SHUT_WR);
}

bool CLTcpConn::shutDownRead(void) 
{
	return m_socket->shutDown(CL_SHUT_RD);
}

bool CLTcpConn::shutDown(void) 
{
	return m_socket->shutDown(CL_SHUT_RDWR);
}

void CLTcpConn::handleEstablished(void) 
{
  assert(m_curState == TCP_CONNECTING);
  m_ioChannel->enableReading(); 
  m_ioLooper->addChannel(m_ioChannel.get());
  setState(TCP_CONNECTED); 
  if(m_connCallback)
  {
    m_connCallback(shared_from_this());
  }
	CLLOG_D("NET", "%s", "CLTcpConn established");
}

void CLTcpConn::handleDestroy(void)  
{
  assert(m_ioLooper->isInLooperThread());
  if(m_curState == TCP_CONNECTED)
  {
    m_ioChannel->disableAll();
    m_ioLooper->removeChannel(m_ioChannel.get());
    setState(TCP_DISCONNECTED);
	  CLLOG_D("NET", "%s", "CLTcpConn destroyed");
  }
}

bool CLTcpConn::send(const string& msg)
{
  return send(msg.data(), msg.size());
}

bool CLTcpConn::send(const void* data, size_t siz)
{
  if(m_curState == TCP_CONNECTED)
	{
    m_ioLooper->runJob(bind(&CLTcpConn::sendInLooper, this, data, siz));
	  return true;
	}
	return false;
}

void CLTcpConn::sendInLooper(const void* data, size_t siz)
{
  assert(m_ioLooper->isInLooperThread());
  CLLOG_D("NET", "CLTcpConn is going to write %u bytes", siz);
  ssize_t wbytes = m_socket->send(data, siz);
	if((static_cast<size_t>(wbytes) == siz) && m_writeCompleteCallback)
	{
    m_ioLooper->queueJob(bind(m_writeCompleteCallback, shared_from_this()));
	}
	else if(!m_ioChannel->isWriteEnabled() && (m_socket->hasDataToSend()))
	{
    // If we did not write all data, enable write for the socket. 
	  m_ioChannel->enableWriting();
		m_ioLooper->modifyChannel(m_ioChannel.get());
	  CLLOG_D("NET", "%s", "CLTcpConn has more data to send");
	}
	CLLOG_D("NET", "CLTcpConn write %d bytes", wbytes);
}

void CLTcpConn::handleRead(void) 
{
  assert(m_ioLooper->isInLooperThread());
  ssize_t rbytes = m_socket->receive();
  if(rbytes > 0)
  {
	  CLLOG_D("NET", "CLTcpConn::handleRead receive %d bytes", rbytes);
    if(m_msgCallback)
    {
      m_msgCallback(shared_from_this(), m_socket->readerBuf(), CLTimestamp::nowTime());
    }
  }
  else if(rbytes == 0)
  {
    handleClose();
  }
  else
  {
    handleError();
  }
}

void CLTcpConn::handleWrite(void) 
{
	assert(m_ioLooper->isInLooperThread());
  assert(m_ioChannel->isWriteEnabled()); 
  ssize_t wbytes = m_socket->sendOutLeft();
  if(wbytes > 0)
	{
    if(!m_socket->hasDataToSend())
		{
      m_ioChannel->disableWriting();
		  m_ioLooper->modifyChannel(m_ioChannel.get());
	    if(m_writeCompleteCallback)
			{
        m_ioLooper->queueJob(bind(m_writeCompleteCallback, shared_from_this()));
			}
		}
	}
	CLLOG_D("NET", "CLTcpConn::handleWrite write %d more bytes", wbytes);
}

void CLTcpConn::handleClose(void) 
{
  if(m_curState == TCP_CONNECTED)
  {
    m_ioChannel->disableAll();
    m_ioLooper->removeChannel(m_ioChannel.get());
    if(m_closeCallback)
    {
      m_closeCallback(shared_from_this());
    }
    setState(TCP_DISCONNECTED);
  }
	CLLOG_D("NET", "CLTcpConn closed: %d", m_ioChannel->fd());
}

void CLTcpConn::handleError(void) 
{
  char errmsg[512];
  int errcode = m_socket->lastErrorCode();
  CLLOG_C("NET", "CLTcpConn::handleError: %d %s", 
			errcode, strerror_r(errcode, errmsg, sizeof(errmsg)));
}

int CLTcpConn::connId(void) const
{
	return m_connId;
}

CLEventLooper* CLTcpConn::ioLooper(void) 
{
	return m_ioLooper;
}

void CLTcpConn::setConnCallback(const CLConnCallback& cb) 
{
  m_connCallback = cb;
}

void CLTcpConn::setMessageCallback(const CLMessageCallback& cb) 
{
  m_msgCallback = cb;
}

void CLTcpConn::setWriteCompleteCallback(const CLWriteCompleteCallback& cb) 
{
  m_writeCompleteCallback = cb;
}

void CLTcpConn::setCloseCallback(const CLCloseCallback& cb) 
{
  m_closeCallback = cb;
}

void CLTcpConn::setState(CLTcpConn::CLTcpState st)
{
  m_curState = st;
}

} // neo
